package MicroTime;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;

public class PhoenixSink extends RichSinkFunction<String> {
    
    private Connection conn;
    
    @Override
    public void open(Configuration parameters) throws Exception {
        // 建立到Phoenix的连接
        String url = Constants.PHOENIX_URL;
        conn = JdbcUtil.getPhoenixConnection(url);
    }
    
    @Override
    public void close() throws Exception {
        // 关闭连接
        if (conn != null) {
            conn.close();
        }
        
    }
    
    @Override
    public void invoke(String value,
                       Context ctx) throws Exception {
        writeData(value);
    }

    private void writeData(String value) throws SQLException {
        JSONObject jsonObject = JSON.parseObject(value);

        // 拼接插入语句 todo
        // upsert into t(a, b, c) values(?, ?, ?)
        StringBuilder sql = new StringBuilder();
        sql
            .append("upsert into ")
            .append(Constants.PHOENIX_TABLE_NAME)
            .append(" (")
            .append(Constants.MICRO_ON_TIME_COLUMN)
            .append(") values (")
            .append(Constants.MICRO_ON_TIME_COLUMN.replaceAll("[^,]+", "?"))
            .append(")");
        System.out.println(sql);
     PreparedStatement ps = conn.prepareStatement(sql.toString());
        // 给占位符赋值 todo  根据列名去data中取值
        String[] cs = Constants.MICRO_ON_TIME_COLUMNS_JSON.split(",");
        for (int i = 0; i < cs.length; i++) {
            String columnName = cs[i];
            Object v = jsonObject.get(columnName);
            System.out.println(i+1+"==>"+v);
           ps.setString(i + 1, v == null ? null : v.toString());
        }

        // 执行sql
        ps.execute();
        conn.commit();
        ps.close();
    }
}